package tsif.tcluster.rpc;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.zip.GZIPInputStream;
import java.util.zip.GZIPOutputStream;

import tsif.tcluster.TClusterMaster.ContextNode;
import tsif.tcluster.netty.codec.TClusterNettyMessage;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;

/**
 * 集群云Rpc工具<br>
 * 用于实现rpc的序列化操作(基于Kryo实现)<br>
 * 
 * 
 */
public class TClusterRpcUtils {
	protected static Set<Class<?>> registers = new LinkedHashSet<>(); // 解析注册类(一定需要有序的)

	static {
		TClusterRpcUtils.addRegisterClass(HashMap.class);
		TClusterRpcUtils.addRegisterClass(ArrayList.class);

		TClusterRpcUtils.addRegisterClass(TClusterNettyMessage.class);
		TClusterRpcUtils.addRegisterClass(TClusterRpcCallMsg.class);
		TClusterRpcUtils.addRegisterClass(TClusterRpcCallBackMsg.class);
		TClusterRpcUtils.addRegisterClass(TClusterRpcInfoMsg.class);

		TClusterRpcUtils.addRegisterClass(ContextNode.class); // 添加了rpc会报错
	}

	public static void addRegisterClass(Class<?> clazz) {
		registers.add(clazz);
	}

	/** 序列化工具，线程安全 **/
	protected static final ThreadLocal<Kryo> kryos = new ThreadLocal<Kryo>() {
		protected Kryo initialValue() {
			Kryo kryo = new Kryo();
			// 遍历添加
			for (Class<?> register : registers) {
				kryo.register(register);
			}
			return kryo;
		};
	};

	/** 获取Kryo **/
	protected static Kryo getKryo() {
		return kryos.get();
	}

	/** 把对象转成数组 **/
	public static byte[] toByte(Object obj, Class<?> clazz) {
		if (obj == null) {
			return null;
		}

		// 创建消息
		Kryo kryo = getKryo();
		ByteArrayOutputStream stream = new ByteArrayOutputStream();
		Output output = new Output(stream);
		kryo.writeObjectOrNull(output, obj, clazz);
		output.flush();
		// 输出结果
		byte[] data = stream.toByteArray();
		return data;
	}

	/** 转化对象 **/
	public static <T> T toObject(byte[] data, Class<T> clazz) {
		if (data == null) {
			return null;
		}
		Kryo kryo = getKryo();
		Input input = new Input(data);
		return kryo.readObjectOrNull(input, clazz);
	}

	/** 达成字节流格式, 自动处理类转化 **/
	public static byte[] toAutoByte(Object obj) {
		if (obj == null) {
			return null;
		}

		// 创建消息
		Kryo kryo = getKryo();
		ByteArrayOutputStream stream = new ByteArrayOutputStream();
		Output output = new Output(stream);
		kryo.writeClassAndObject(output, obj);
		output.flush();
		// 输出结果
		byte[] data = stream.toByteArray();
		return data;
	}

	public static Object toAutoObject(byte[] data) {
		if (data == null) {
			return null;
		}
		Kryo kryo = getKryo();
		Input input = new Input(data);
		return kryo.readClassAndObject(input);
	}

	/** 使用gzip进行压缩 */
	public static byte[] gzip(byte[] src) {
		// 检测数据
		if (src == null) {
			return null;
		}

		ByteArrayOutputStream out = new ByteArrayOutputStream();
		GZIPOutputStream gzip = null;
		try {
			gzip = new GZIPOutputStream(out);
			gzip.write(src);
			gzip.finish();
			return out.toByteArray();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (gzip != null) {
				try {
					gzip.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return null;
	}

	/** gzip解压 **/
	public static byte[] ungzip(byte[] compressed) {
		// 检测数据
		if (compressed == null) {
			return null;
		}

		ByteArrayOutputStream out = new ByteArrayOutputStream();
		ByteArrayInputStream in = new ByteArrayInputStream(compressed);
		GZIPInputStream gzip = null;
		try {
			gzip = new GZIPInputStream(in);
			// 遍历读取
			byte[] data = new byte[1024];
			while (true) {
				// 读取数据
				int gsize = gzip.read(data);
				if (gsize <= 0) {
					break;
				}
				// 写入数据
				out.write(data, 0, gsize);
			}

			return out.toByteArray();
		} catch (IOException e) {
			e.printStackTrace();
		} finally {
			if (out != null) {
				try {
					out.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
			if (gzip != null) {
				try {
					gzip.close();
				} catch (IOException e) {
					e.printStackTrace();
				}
			}
		}
		return null;
	}
}
